package zju.dgm.dto;

import lombok.Data;
import lombok.ToString;
import zju.dgm.util.SyncUtil;

import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 记录查询次数,用于负载均衡
 *
 * @author lzy
 */
@Data
public class QueryStat implements Serializable {


    ConcurrentHashMap<String, PriorityQueue<RegionEntry>> tableStats = new ConcurrentHashMap<>();


    public QueryStat() {
    }

    public static PriorityQueue<RegionEntry> minHeapInstance() {
        return new PriorityQueue<RegionEntry>((o1, o2) -> {
            if (o1.getQTimes() != o2.getQTimes()) {
                return o1.getQTimes() - o2.getQTimes();
            } else if (o1.getServer().getPort() != o2.getServer().getPort()) {
                return o1.getServer().getPort() - o2.getServer().getPort();
            } else {
                return 1;
            }
        });
    }


    public ServerInfo getTablePos(String tableName) throws Exception {
        assert tableStats.containsKey(tableName);
        PriorityQueue<RegionEntry> minHeap = tableStats.get(tableName);
        assert minHeap.size() > 0;

        synchronized (this) {
            RegionEntry rst = minHeap.poll();
            if (rst == null) {
                throw new NoSuchElementException("no such server");
            }
            rst.qTimes++;
            minHeap.add(rst);
            return rst.server;
        }
    }

    public void updateStats(ConcurrentHashMap<String, PriorityQueue<RegionEntry>> newStats) {
        this.tableStats = newStats;
    }

    /**
     * 根据从Region那里查到的map，更新自身的regionInfo
     *
     * @param newMap
     */
    public synchronized void syncInfoIncrement(Map<String, List<ServerInfo>> newMap) {
        newMap.forEach((k, v) -> {
            if (!tableStats.containsKey(k)) {// 有新表加入
                PriorityQueue<RegionEntry> newHeap = minHeapInstance();
                for (ServerInfo serverInfo : v) {
                    newHeap.add(new RegionEntry(serverInfo, 0));
                }
                tableStats.putIfAbsent(k, newHeap);
            }

            PriorityQueue<RegionEntry> heap = tableStats.get(k);
            if (v.size() > heap.size()) {// 有server新增，表对应的server数增加了
                HashMap<ServerInfo, Integer> tempMap = new HashMap<>();
                for (RegionEntry regionEntry : heap) {
                    tempMap.put(regionEntry.getServer(), regionEntry.getQTimes());
                }
                for (ServerInfo serverInfo : v) {
                    if (!tempMap.containsKey(serverInfo)) {// 找到了新加的server
                        heap.add(new RegionEntry(serverInfo, 0));
                    }
                }
            } else if (v.size() < heap.size()) {// 有server削减
                for (RegionEntry regionEntry : heap) {
                    if (!v.contains(regionEntry.getServer())) {// 找到了被削减的server
                        heap.remove(regionEntry);
                    }
                }
            }
        });
    }

    /**
     * 默认初始化
     *
     * @param regionMap
     */
    public void defaultInit(Map<String, List<ServerInfo>> regionMap) {
        tableStats.clear();
        if (regionMap == null || regionMap.isEmpty()) {
            tableStats = new ConcurrentHashMap<>();
            return;
        }

        for (String tabName : regionMap.keySet()) {
            PriorityQueue<RegionEntry> minHeap = minHeapInstance();
            tableStats.put(tabName, minHeap);
            for (ServerInfo serverInfo : regionMap.get(tabName)) {
                minHeap.add(new RegionEntry(serverInfo, 0));
            }
        }
    }

    public ConcurrentHashMap<String, PriorityQueue<RegionEntry>> getTableStats() {
        return this.tableStats;
    }

    synchronized public void touch(String table) {
        PriorityQueue<RegionEntry> heap = tableStats.get(table);
        RegionEntry cp = null;
        if (heap == null) {
            return;
        }
        for (RegionEntry regionEntry : heap) {
            if (regionEntry.getServer().equals(SyncUtil.serverInfo)) {
                cp = regionEntry;
                break;
            }
        }
        if (cp != null && heap.remove(cp)) {
            cp.increment(1);
            heap.add(cp);
        }
    }
}


